
#include "bulletinboard.h"
#include "config.h"
#include "constants.h"
#include "replica.h"
#include "utils.h"
#include "writebuffer.h"

#include <errno.h>
#include <fcntl.h>
#include <getopt.h>
#include <netdb.h>
#include <poll.h>
#include <pthread.h>
#include <syslog.h>
#include <sys/ioctl.h>
#include <sys/socket.h>

#include <cstdio>
#include <cstdlib>
#include <cstring>
#include <ctime>
#include <iostream>
#include <map>
#include <set>
#include <string>
#include <signal.h>

using namespace std;


// Server exit codes used for consistency.
enum ExitCode
{
    NoError = 0,
    CommandLineError,
    ForkError,
    ThreadError,
    SocketError
};


// Global Variables

// Flags when we're running in debug mode.
bool debugMode;

// Our shared BulletinBoard instance that will do all the actual file access "work".
BulletinBoard bb;

// Data structures used by the concurrency managment algorithm.
pthread_mutex_t threadUsageMutex; // Control access to thread management
map<pthread_t,bool> clientThreads; // Contains all client threads and if they are busy
set<pthread_t> clientThreadsToKill; // List threads that should terminate themselves

// Controls access to the master socket to prevent poll/accept conflicts among
// client threads.
pthread_mutex_t masterSocketMutex;

// Controls access to the shutdown stage.
pthread_mutex_t shutdownStageMutex;
// Enforces proper sequencing in the server shutdown/restart process.
int shutdownStage;
// Stores the last handled signal.
int receivedSignal;


// Simple signal handler to store the signal for later use.
void signalHandler( int signal )
{
    receivedSignal = signal;
}


/* This is the function executed by the threads that will handle connection directly with client
 * and satisfy its requests.
 */
void *client_thread(void *ms)
{
    int msock = (intptr_t) ms; //Master socket

    log( LOG_DEBUG, "New client thread created.\n" );

    // Loop forever.
    while ( true )
    {
        struct pollfd fds;
        fds.revents = 0;
        fds.events = POLLIN;
        fds.fd = msock;

        // Only one client thread may access the master socket simultaneously.
        // The first to acquire this lock gets a chance to poll (with timeout)
        // and accept any new connections, but it must then release it to give
        // others a chance.
        pthread_mutex_lock(&masterSocketMutex);
        int sd = -1;
        int poll_result = poll(&fds, (nfds_t) 1, pollTimeout);
        if (poll_result > 0)
        {
            struct sockaddr_in client_addr; //The address of the client
            socklen_t client_addr_len = sizeof(client_addr); //Client's length
            sd = accept(msock, (struct sockaddr*)&client_addr, &client_addr_len); //If there is a pending connection, accepts it.
        }
        pthread_mutex_unlock(&masterSocketMutex);

        // If we successfully created a client connection...
        if ( sd >= 0 )
        {
            log( LOG_DEBUG, "New client arrived.\n" );

            // Mark the thread as busy.
            pthread_mutex_lock( &threadUsageMutex );
            clientThreads[ pthread_self() ] = true;
            pthread_mutex_unlock( &threadUsageMutex ); //Now that the thread is not in the inactive thread pool

            // Initialise the username to the default "nobody".
            string user = defaultUserName;

            // Read lines from the socket and act on them.
            bool done = false;
            while ( !done )
            {
                string line;
                int bytesRead = readLine( sd, line );

                // On read errors we drop the client. I'm not sure this is the
                // best course of action, but it is the easiest.
                if ( bytesRead == -1 )
                {
                    logError(LOG_WARNING, "Read from slave socket");
                    break;
                }

                // Zero bytes read means the client closed the connection, so
                // move on.
                if ( bytesRead == 0 )
                {
                    string reply = "BYE: It's rude to leave without saying goodbye.\n";
                    send( sd, reply.c_str(), reply.size(), 0 );
                    break;
                }

                // If the server is shutting down, we end the conversation early.
                pthread_mutex_lock( &shutdownStageMutex );
                bool gottaGo = shutdownStage == 2;
                pthread_mutex_unlock( &shutdownStageMutex );
                if ( gottaGo )
                {
                    string reply = "BYE: Sorry, but I have to go. Come back soon.\n";
                    send( sd, reply.c_str(), reply.size(), 0 );
                    break;  
                }

                // Otherwise attempt to execute the given command and send a response.
                string reply = bb.interpretCommand( line, user, done );
                send( sd, reply.c_str(), reply.size(), 0 );
            }

            // Shutdown and close the socket.
            shutdown( sd, SHUT_RDWR );
            close(sd);

            log( LOG_DEBUG, "Client left.\n" );

            // Mark the thread as idle.
            pthread_mutex_lock(&threadUsageMutex);
            clientThreads[pthread_self()] = false;
            pthread_mutex_unlock(&threadUsageMutex);
        }

        // Check if we're being asked to kill ourselves.
        pthread_mutex_lock( &threadUsageMutex );
        if ( clientThreadsToKill.find( pthread_self() ) != clientThreadsToKill.end() )
        {
            clientThreads.erase( pthread_self() );
            clientThreadsToKill.erase( pthread_self() );
            pthread_mutex_unlock( &threadUsageMutex );
            break;
        }
        pthread_mutex_unlock( &threadUsageMutex );

        pthread_yield();
    }

    log( LOG_DEBUG, "Client thread terminating.\n" );

    return NULL;
}

/* This function monitor all slave threads. It it is needed, it allocates or dealloactes
 * a batch of T_INCR threads.
 * There will always be at least T_INCR threads allocated and never more than T_INCR + T_MAX.
 * If the number of active threads is T_INCR + 1 less than the number of allocated threads,
 * this thread will reap T_INCR threads only if all conditions above are satisfied.
 */
void *thread_monitor(void *ms)
{
    log( LOG_DEBUG, "Monitor thread started.\n" );

    int msock = (intptr_t) ms; //Master socket

    // Get the concurrency parameters from the config.
    int thInc = Config::self()->threadIncrement();
    int thMax = Config::self()->maxThreads();

    /* Initialize attributes so that the threads will be dettached. */
    pthread_attr_t attr;
    pthread_attr_init(&attr);
    pthread_attr_setdetachstate(&attr,PTHREAD_CREATE_DETACHED);

    //Allocates the initial batch of T_INCR threads.
    pthread_mutex_lock(&threadUsageMutex);
    for (int i = 0; i < thInc; i++)
    {
        pthread_t thread;
        if ( pthread_create(&thread, &attr, client_thread, (void*)msock) != 0 )
        {
            logError(LOG_ERR, "Creating client thread");
        }
        else
        {
            clientThreads[ thread ] = false;
        }
    }
    pthread_mutex_unlock(&threadUsageMutex);

    // Loop forever.
    while ( true )
    {
        pthread_mutex_lock(&threadUsageMutex);

        // Determine how many threads are currently active.
        int activeThreadCount = 0;
        for ( map<pthread_t,bool>::const_iterator it = clientThreads.begin(); it != clientThreads.end(); ++it )
            if ( (*it).second )
                ++activeThreadCount;

        // If more threads are needed create some.
        if (activeThreadCount == clientThreads.size()) 
        {
            if ( clientThreads.size() <= thMax )
            {
                for (int i = 0; i < thInc; i++)
                {
                    pthread_t thread;
                    if ( pthread_create(&thread, &attr, client_thread, (void*)msock) != 0 )
                    {
                        logError( LOG_ERR, "Client thread creation");
                    }
                    else
                    {
                        clientThreads[thread] = false; //All threads are initially inactive.
                    }
                }
            }
        }
        // If we have unneeded threads terminate some.
        else if (( clientThreads.size() > thInc) && ( clientThreads.size() - activeThreadCount > thInc))
        {
            // Add thInc idle threads to the set of threads to kill themselves.
            int numThreadsToKill = thInc;
            for ( map<pthread_t,bool>::const_iterator it = clientThreads.begin();
                  numThreadsToKill > 0 && it != clientThreads.end();
                  ++it )
            {
                if ( !(*it).second )
                {
                    clientThreadsToKill.insert( (*it).first );
                    --numThreadsToKill;
                }
            }
        }

        pthread_mutex_unlock(&threadUsageMutex); //Unlocks mutex - other threads now may become active or inactive.

        // If we have received a signal, the monitor thread begins the process
        // of shutting down the server. We stop the concurrency management loop.
        if ( receivedSignal != 0 )
        {
            pthread_mutex_lock( &shutdownStageMutex );
            shutdownStage = 1;
            pthread_mutex_unlock( &shutdownStageMutex );
            break;
        }

        // The monitor thread doesn't need to run continuously, so we sleep
        // for a while.
        sleep( monitorThreadInterval );
    }

    // Tell all threads to kill themselves when they get the chance.
    pthread_mutex_lock( &threadUsageMutex );
    for ( map<pthread_t,bool>::const_iterator it = clientThreads.begin();
         it != clientThreads.end();
         ++it )
    {
        clientThreadsToKill.insert( (*it).first );
    }
    pthread_mutex_unlock( &threadUsageMutex );

    return NULL;
}


int main( int argc, char * * argv )
{
    // Initialise the random number generator.
    srand( time( NULL ) );

    // Initialise the mutices.
    pthread_mutex_init( &threadUsageMutex, NULL );
    pthread_mutex_init( &masterSocketMutex, NULL );
    pthread_mutex_init( &shutdownStageMutex, NULL );


    // Read command line options.
    debugMode = false;
    string configFile = defaultConfigFile;
    string bbFile = defaultBbFile;

    while ( true )
    {
        char flag = getopt( argc, argv, "dc:" );
        if ( flag == -1 )
            break;
        else if ( flag == 'd' )
            debugMode = true;
        else if ( flag == 'c' )
            configFile = optarg;
        else
            return CommandLineError;
    }
    if ( optind < argc )
    {
        bbFile = argv[optind];
        ++optind;
    }
    if ( optind < argc )
    {
        log( LOG_CRIT, "Detected extra command line argument: \"%s\" Aborting.\n", argv[optind] );
        return CommandLineError;
    }

    // If not in debug mode we need to daemonise the server.
    if ( !debugMode )
    {
        // Fork to the background.
        int childPid = fork();
        if ( childPid < 0 )
        {
            perror( "Startup fork" );
            return ForkError;
        }
        else if ( childPid > 0 )
        {
            printf( "Forked. pid = %d", childPid );
            return NoError;
        }

        // Close all file descriptors.
        for ( int i = 0; i < getdtablesize(); i++ )
            close( i );

        // Open /dev/null as standard input.
        open( "/dev/null", O_RDWR );

        // Setup logging
        if ( Config::self()->useSyslog() )
        {
            // Start logging to syslogd if enabled in the config.
            openlog( "bbserv", LOG_CONS | LOG_PID, LOG_USER );
        }
        else
        {
            // Otherwise, redirect stdout and stderr to a log file.
            mode_t mode = S_IRUSR | S_IWUSR | S_IRGRP | S_IROTH;
            int logd = open( Config::self()->logFile().c_str(), O_RDWR | O_CREAT | O_APPEND, mode );
            dup( logd );
        }

        // Detach from TTY.
        int ttyd = open( "/dev/tty", O_RDWR );
        ioctl( ttyd, TIOCNOTTY, 0 );
        close( ttyd );
    }
    else
    {
        // If running in debug mode, shutdown cleanly on ^C.
        signal( SIGINT, signalHandler );
    }

    // Install signal handlers for SIGHUP and SIGQUIT.
    signal( SIGHUP, signalHandler );
    signal( SIGQUIT, signalHandler );

    // Load the bulletinboard file.
    bb.setFileName( bbFile );

    do
    {
        // Reset the signal and shutdown stage.
        receivedSignal = 0;
        shutdownStage = 0;

        // Read the (new?) configuration.
        if ( Config::self()->read( configFile ) )
            log( LOG_INFO, "Problem reading config from %s.\n", configFile.c_str() );

        // Create the socket for incoming client connections.
        int masterSocket = master_socket(Config::self()->bbPort());
        if (masterSocket == -1)
        {
            logError( LOG_CRIT, "Master socket creation" );
            return SocketError;
        }

        pthread_attr_t attr;
        pthread_attr_init(&attr);

        pthread_t monitor; //Monitor thread that will spawn or kill slave threads
        int errn = pthread_create(&monitor, &attr, thread_monitor, (void *)masterSocket); //Creates monitor thread - it spawns and kills slave threads if necessary.
        if (errn)
        {
            logError( LOG_CRIT, "Monitor thread creation" );
            return ThreadError;
        }

        pthread_t replicaThread; //Replica thread that will sync with other servers.
        errn = pthread_create(&replicaThread, &attr, replica_thread, NULL);
        if (errn)
        {
            logError( LOG_CRIT, "Replica thread creation" );
            return ThreadError;
        }



        // Shutting down.

        // Wait for monitor thread to finish.
        pthread_join( monitor, NULL );
        log( LOG_DEBUG, "Monitor thread done.\n" );

        // Increment the shutdown stage. This will cause
        // client threads to abandon their clients.
        pthread_mutex_lock( &shutdownStageMutex );
        shutdownStage = 2;
        pthread_mutex_unlock( &shutdownStageMutex );

        // Shutdown and clost the master socket.
        shutdown( masterSocket, SHUT_RDWR );
        close( masterSocket );

        // Empty the queue of messages waiting to be written.
        while ( WriteQueue::self()->hasMessageWaiting() )
        {
            WriteQueue::self()->doneWithMessage( -6 );
        }

        // Wait for all client threads to terminate.
        while ( true )
        {
            sleep( 1 );
            pthread_mutex_lock( &threadUsageMutex );
            bool clientsDone = clientThreadsToKill.empty();
            pthread_mutex_unlock( &threadUsageMutex );
            if ( clientsDone )
                break;
        }
        log( LOG_DEBUG, "Client threads done.\n");

        // Increment the shutdown stage. This will cause the replica thread
        // to begin shutting down.
        pthread_mutex_lock( &shutdownStageMutex );
        shutdownStage = 3;
        pthread_mutex_unlock( &shutdownStageMutex );

        // Wait for replica thread to terminate.
        pthread_join( replicaThread, NULL );
        log( LOG_DEBUG, "Replica thread done.\n");
    }
    while ( receivedSignal == SIGHUP ); // If SIGHUP, restart. Otherwise, we're done.

    log( LOG_INFO, "Server shutting down.\n" );
    closelog();

    return NoError;
}
